Bound how long the publisher lease can go unrenewed: a deadline for every request under it - #159

A lease renewal is three ClickHouse requests -- the head read, the claim
INSERT, the read-back -- and each was bounded only by the catalog client's
request timeout, 60 s by default against a 15 s TTL, while the lease thread
held the lease lock. Every other request made under that lock had the same
bound: an index pass's version claim, descriptor INSERTs, publish and
read-backs, a reconcile's replay-guard read. A catalog that accepted a
request and never answered let the row expire with the service still
reporting the lease held, and a rival could take the catalog before any
error surfaced. The claim INSERT also carried no server-side cap, so the
server could land it after the client had given up.

A first attempt (never pushed) gave each lease statement a fixed
(TTL / 3 - skew) / 4, 1.25 s at the defaults. Review found that it held only
while the service was idle -- the index pass's own requests kept their 60 s
under the lock -- and that it quarantined a slow but healthy catalog (a
select_sequential_consistency read, a quorum INSERT whose wait it cut from
5 s to 1.25 s), failed start() outright on a cold server, and could go round
silently: claim times out, quarantine, refused by its own late row, repeat,
saying only "curl: Timeout was reached". This replaces it.

The deadline. The server stamps a lease row no earlier than its claim
INSERT was sent, and a rival whose clock runs clock_skew ahead sees it
expire that much early, so on the claimant's steady clock the row keeps
rivals out until

  lease deadline = claim INSERT sent + lease_ttl - clock_skew - 0.1 s

the 0.1 s covering the time between a request failing and the writer
saying so. Every request made under the lease has to be answered by then,
however the time is spread across them, so one slow but healthy request may
use all of it.

- ClickHouseClient: a RequestDeadline is a thread-local, nestable scope
  holding an absolute steady-clock deadline, read afresh before every
  attempt, so a renewal inside the scope extends it at once. Each attempt's
  timeouts are cut to the whole milliseconds left; a request whose deadline
  has passed is not sent; #151's retries honour it -- no attempt, and no
  backoff sleep, past the deadline. ClickHouseError now says whether the
  request timed out and whether it can have reached the server (sent()),
  and a timeout names the bound that ended it and the knobs behind it.
- LeaseCoordinator: its statements run under the held lease's deadline
  while that is still ahead. A claim made without a lease -- at start, or
  after a quarantine -- gets min(request timeout, TTL / 3) per request, and
  so does one made under a lease whose deadline has passed: the protocol's
  own reads still decide such a claim (a renewal that meets a successor is
  refused, a publish is fenced out), where refusing to send it would turn
  those known outcomes into unknown ones. The lease INSERTs carry the time
  left as max_execution_time and lock_acquire_timeout (whole seconds rounded
  down, a fraction under one; both accepted as URL settings by 25.12), with
  timeout_overflow_mode=throw, and insert_quorum_timeout becomes
  min(publish_timeout, time left). The statement text is unchanged, and
  tests/test_native_catalog_lease_live.py passes unmodified.
- CaptureStorageService: every use of the lease goes through a LeaseScope,
  which takes the lease lock and installs the lease deadline for every
  request made under it; a lease whose deadline has passed is abandoned
  (quarantined, no tombstone) on the way in and out, so it is never used or
  reported held. The renewal schedule is unchanged (a third of the TTL after
  the claim that stamped the row was sent).
- Slow catalogs. CatalogWriter::acquire_lease() quarantines only when the
  claim's INSERT can have reached the server: a claim whose head read timed
  out, or could not connect, wrote nothing, and is retried a lease tick
  later instead of a TTL. This deliberately differs from the Python oracle
  (acquire_publisher_lease), which quarantines on any error. start() retries
  a claim that timed out within start_lease_wait, waiting out a quarantine
  when it ends inside the wait. Lease claims and renewals that time out are
  counted until one succeeds (snapshot lease_timeouts); from the third,
  lease_timeout_error -- and last_error -- names lease_ttl_s, clock_skew_s,
  clickhouse_request_timeout_s and lease_ttl_s / 3 with their values.

The service constructor and NativeCaptureStorageConfig require the renewal
window -- TTL / 2, when a renewal starts at the latest, to the deadline --
to be at least 200 ms: clock_skew <= TTL / 2 - 0.3 s. Main checked only the
fence margin (TTL - publish_timeout - skew >= 0.1 s), so this newly refuses
a skew above TTL / 2 - 0.3 s that the fence admits -- above 7.2 s and up to
9.9 s at the defaults (TTL 15 s, publish 5 s) -- and newly accepts nothing.
Against the first attempt's rule (skew <= TTL / 3 - 0.2 s) it accepts skews
from there up to TTL / 2 - 0.3 s, 5 s at the defaults for one. The defaults
and every configuration the suites use pass. docs/integration-api-v1.md
states the bound and the rule. tests/tools/verify_replicated_quorum.py (its
Keeper harness is gone, so not run) keeps expecting insert_quorum_timeout
5000 on the lease INSERTs: at its 30 s TTL the time left is 10 s or more.

Not covered here: an index pass still reads its packs from the object store
under the lease lock (the next commit).

Tests first, each red on main and, where it guards against the first
attempt, on that design rebuilt on #151's client:

- tests/test_native_lease_request_bound.py (CPU; conformance_catalog against
  a scripted HTTP server): the lease INSERTs carry the time left as
  max_execution_time and lock_acquire_timeout (main: no cap; first attempt:
  TTL / 12, no lock_acquire_timeout); the quorum wait is capped by it (main:
  publish_timeout's 5000 ms against 4 s left); a stalled head read, INSERT or
  read-back of a claim fails within TTL / 3, naming the knobs (main: not
  bounded; first attempt: 250 ms, naming nothing); a stalled renewal fails
  0.1 s before the row expires (first attempt: after 250 ms); retries stop at
  the deadline (main: all three 503 attempts, 1.5 s); only a claim that sent
  its INSERT quarantines (first attempt: the head-read case did too); a 2 s
  head read no longer fails the claim (first attempt: 1.25 s timeout); and
  the skew rule both ways for the native service, and for the Python config
  in tests/test_native_capture_storage_wiring.py.
- tests/test_native_capture_storage_live.py, 3 s TTL and a 20 s request
  timeout through a TCP switch that can now stall only the requests a
  predicate picks, or hold one back: a stalled renewal (main: "held" over a
  dead row from 3.06 s); every catalog INSERT but the lease's stalled inside
  an index pass, the review's repro (main: "held" over a dead row from
  2.99 s; first attempt: from 2.94 s) -- now quarantined before 3 s with the
  row still live; start() with its first lease read 1.5 s late (first
  attempt: start() failed with the 250 ms timeout) -- now held after one
  retry, well under the 3 s quarantine; a catalog that stops answering
  surfaces lease_timeout_error naming the knobs, cleared once the lease
  renews (both: no such field); and query_log shows the caps on the lease
  INSERTs. The switch now answers libcurl's Expect: 100-continue itself, so
  routing a statement over 1 KiB no longer adds a second.

At this commit the CPU lease and wiring tests and the storage and lease
live suites pass.

The index pass held the lease lock across the whole of indexer_.index(),
its object-store reads included: each pending pack's footer and descriptor
rows come from ranged GETs, bounded only by the S3 client's read timeout,
120 s by default. The lease deadline bounds catalog requests, not those, so
a GET that stalled kept the lease thread from renewing for as long, and the
row expired while the snapshot said "held".

NativeIndexer::index() is now plan() -- deduplication and the replay guard,
a catalog read -- then read(), the object store only, then commit(): the
version, the descriptor writes, the publish and the inventory. The storage
service holds the lease lock (a LeaseScope) for plan() and for commit() and
reads the packs without it, so the lease keeps renewing through a stalled
read. index() still runs the three in one call, for the conformance driver.

Two things follow from letting go of the lock mid-pass:

- commit() may run under another lease than plan() did: the lease can be
  lost and taken afresh while the pass reads, and another publisher can hold
  the catalog meanwhile and index the same packs, which its reconcile finds
  uploaded and uncommitted. The plan records the lease_id its replay guard
  was read under, and commit() reads the guard again when the lease is not
  that one, dropping what has been committed since. Trusting the old read
  published those packs a second time, at a higher version.
- a long commit (many descriptor chunks) must not run the lease down while
  it holds the lock. IndexerConfig::keep_lease is called before the version
  claim and before each descriptor chunk, and the service renews there when
  the renewal is due. Not before the inventory commit: that follows a
  publish, which has just renewed, and a renewal failing there would leave
  the packs it made visible unrecorded.

Tests first, in tests/test_native_capture_storage_live.py. The TCP switch
can now stand in front of the fake S3 too. With every ranged GET stalled
(for a new pack only the indexer's footer reads are ranged), the lease must
stay held and live for 6 s, two TTLs, renewing at least three times. On
the previous commit the snapshot said "held" over a dead row from 2.92 s
into the stall, and main failed it too. The pack indexes once the read
fails.

A conformance-driver flag, rival_indexes_before_commit, releases the lease
between read() and commit(), lets a second writer take it and index the
same packs, and takes a fresh lease: with the re-read disabled the pass
published both packs again (indexed_packs 2); with it both are skipped, and
the catalog holds each descriptor once, at the rival's version.

At this commit the CPU lease tests and the storage and lease live suites
pass.

The caps on a lease INSERT cannot cover every late landing: ClickHouse
checks max_execution_time only at points inside the pipeline, and starts its
clock when the query reaches the server, so a claim held up in transit still
lands after the client gave up and quarantined. That row then refuses the
service's next claim once the quarantine ends. The refusal clock is reset
only when a claim succeeds, so refusals by a rival that had since left,
followed by refusals from the service's own late row, added up to 2 x TTL
and latched "held by another publisher" -- naming the service's own holder.

LeaseCoordinator now remembers the lease_ids of its recent claim INSERTs
(sent, confirmed or not; each id once, so the renewals of one lease, which
all claim the same id, cannot push an earlier claim out of the 16-entry
history) and records whether a refusal came only from those rows;
CatalogWriter::refused_by_own_claims() exposes it. The service restarts the
refusal clock on such a refusal instead of counting it. That is sound: the
late claim was only sent because its head read found no live rival, so any
earlier rival's refusals ended there; the row expires one TTL after it
landed; and while it refuses, the service's claims stop at the head read, so
no further rows of its own appear. A contested head with any foreign
claimant still counts. The conformance driver reports the attribution on a
refusal as own_claims.

Test first in tests/test_native_capture_storage_live.py: the TCP switch can
now hold lease INSERTs back and deliver them late. A rival takes over during
a cut and leaves 4 s into its refusals; the first service's next claim
reaches the server 1.5 s late, after the 1 s bound on a claim made without a
lease (lease_ttl_s / 3), and lands inside the quarantine. With the reset
disabled the service latched against 'first-publisher' itself; now it takes
the lease back. The attribution is pinned on the CPU gate in
tests/test_native_lease_request_bound.py (own row: true; contested with a
rival, or another coordinator: false), and so is the history: twenty
renewals of one lease used to crowd out an earlier claim, which then read as
a rival's (own_claims false).

The cpu tier, and the storage, lease, chain, reader-parity and lease-bound
suites live, pass.

start() took the lease, swept the spool and reconciled the bucket, and
only then started the lease thread, so nothing renewed the lease while a
large bucket was listed: listing pages and their committed_pack_ids reads
never renew, and the indexer's keep_lease hook runs only once a missing
pack is being committed. Once the lease deadline passed, the next
lease-locked step abandoned the lease (LeaseScope), and a crash-window
pack found after that hit commit()'s "holds no publisher lease", which
start() rethrew -- where main re-claimed its own lapsed row and indexed
the pack. With nothing to index, start() returned with the lease
quarantined, and the snapshot said "held" over a dead row all through the
listing. stop() had the same gap at the other end: it woke the loop and
the lease thread together, and the lease thread left while the loop's last
cycle still ran.

- start() starts the lease thread as soon as it holds the lease, before
  the sweep and the reconcile (sweep_and_reconcile_at_start). A start
  that fails after that stops the lease thread before releasing.
- stop() joins the loop first and only then the lease thread, which now
  has a stop flag of its own.
- A lease lost while the start-time reconcile runs -- to a quarantine or
  to another holder -- no longer fails start(). That is the running
  service's case, and the loop handles it as it does there: a fresh lease
  once the quarantine ends, or the latch after 2 x TTL of a rival. The
  pass it cut short is owed (reconcile_owed_), and the loop runs it once
  it holds a lease again. A reconcile that fails otherwise (a listing
  error) is recorded as before.

Tests first, in tests/test_native_capture_storage_live.py, TTL 3 s:

- the first LIST delayed 6 s at start, with and without a crash-window
  pack: before, start() raised "holds no publisher lease" (with the pack)
  or returned quarantined (without), and the snapshot said "held" over a
  dead row from 2.9 s; now the lease is held and renewed through the
  listing and the pack is reconciled and indexed.
- the lease thread's renewal stalled while a 2.5 s listing runs, so the
  lease quarantines before the crash-window pack is found: with start()
  rethrowing it raised; now start() returns quarantined, and the loop
  takes a fresh lease and indexes the pack.
- stop() while the loop's last cycle reads a pack for 4 s: with the old
  stop order, "held" over a dead row from 2.97 s; now the cycle indexes
  the pack and the lease is released after it.

The storage live suite and the CPU lease-bound tests pass.

Every request made under the lease lock has to be answered by the lease
deadline, and a pass could renew only at the indexer's keep_lease hook,
before its catalog writes. Between two hooks ran several requests with no
chance to renew: allocate_version's two max() reads, its claim INSERT and
read-back (and last_published_version on a first pass); the publish's
watermark INSERT, owners and member-count reads and the inventory commit.
The stretch and the renewal after it had to fit before the deadline, so
against a catalog slow but healthy -- each request well inside
publish_timeout -- the renewal at the next hook started with too little
time left, timed out, and quarantined the writer. At the defaults with 3 s
INSERTs and 1 s reads: a renewal leaves the lease 4 s old, allocate_version
adds 6 s, and the renewal due at the next hook needs 5 s of the 4.9 s left.
Every pass failed that way, so flush() never drained; main drained it,
bounding nothing.

- RequestDeadline can carry a before_request hook, which execute() runs
  once per call before it reads the deadline, so the hook may move it.
  Only the innermost scope's hook runs, and never inside itself, so the
  lease coordinator's own scope keeps the requests of a renewal the hook
  started from starting another.
- LeaseScope's hook is keep_lease_in_pass(): renew when due (a third of
  the TTL after the claim that stamped the row was sent), before every
  request a stretch under the lease lock sends, reads included, the
  inventory commit too. A renewal then starts with the lease at most a
  third of the TTL plus one request old. IndexerConfig::keep_lease, which
  did this before catalog writes only, is gone.
- The lease thread wakes when the renewal falls due (a tick at most), not
  on a fixed tick, so the renewal a pass leaves due when it lets go of the
  lock starts then, not up to a sixth of the TTL later.

What cannot be kept: a catalog so slow that one request plus a renewal
does not fit in what a renewal leaves -- at a 15 s TTL, 4 s INSERTs with
1 s reads -- still loses the lease pass after pass, now while its row is
still live rather than drained over a dead one as main did; it needs a
longer lease_ttl_s (the next commit makes the service say so).

Test first, in tests/test_native_capture_storage_live.py: the TCP switch
can now hold every request back by a delay a function picks, and a
catalog answering INSERTs 0.6 s late and reads 0.2 s late against a 3 s
TTL (3 s and 1 s at the default 15 s) indexes two packs with the lease
held throughout. Before: flush() timed out with both packs owed, "index
failed: ... Timeout ... bounded by the lease deadline".

The storage and catalog-lease live suites and the CPU lease-bound tests
pass.

The lease INSERTs carry the time left before their deadline to the
server as max_execution_time and lock_acquire_timeout, so the server
abandons a claim when the client does. Two things made the cap wrong.

The caps were computed once, before execute(), and execute() wrote them
into a URL it built once, ahead of the retry loop. A lease INSERT whose
first connections were refused -- retried, since nothing reached the
server -- went out on a later attempt still claiming the time left
before the first: with three attempts, about 0.3 s more than the client
would wait. The URL is now built per attempt, and a per-attempt settings
callback gives the coordinator the time each attempt has (the request
timeout cut to the deadline in force) to cap it with.

And the caps were whole seconds, rounded down: 4 s against the 5 s a
default claim may take, 1 s against 1.999 s. The server aborted healthy
INSERTs the client was still waiting for. They now go to the
millisecond ("4.999"), which the server parses for both settings (checked
on 25.12); a cap under a second already went out as a fraction.

The driver-level tests put a forwarder in front of the fake catalog that
stops listening after the head read is answered, so the INSERT goes out
on its third attempt, and check that its cap ends by the client's
deadline. A second test pins that an INSERT whose every attempt was
refused wrote nothing, so the writer is not quarantined.

A claim made without a lease bounded each of its requests by the claim
bound, min(request timeout, TTL / 3), from when that request started, so
its read-back had a full bound of its own after the INSERT. The lease the
claim takes has until sent + TTL - skew - 0.1 s. Once the clock skew is
over TTL / 3 - 0.1 s -- which validation accepts, up to TTL / 2 - 0.3 s --
an INSERT and a read-back each inside the bound could confirm the claim
after its own deadline. The service counted that as a success (a
reacquisition, "held", the timeout streak reset), then abandoned the
lease at once for "not renewed by the lease deadline". No timeout was
counted and the knobs were never named, so against such a catalog it
looped claim, abandon, quarantine without saying why.

From its INSERT on, a claim now runs under the deadline of the lease it
would take as well as under its own bound: the INSERT, a contested
claim's rival row, the read-back and the head read after a claim not
recorded. A claim that cannot be confirmed in time fails as a timeout
naming lease_ttl_s and clock_skew_s, and, its INSERT sent, quarantines.
On a renewal the old lease's deadline is earlier and still what binds.

start() still failed when its claim INSERT, rather than its head read,
was slow -- the case the claim bound was meant not to fail -- in two
ways.

The client timed out the INSERT. It may have landed, so the writer
quarantined for a TTL, and start() waited that out only when it ended
inside start_lease_wait_s. At the default wait, lease_ttl_s +
publish_timeout_s + clock_skew_s, it never did once the INSERT had used
its claim bound (TTL / 3, the publish timeout at the defaults), so start()
threw "leaves no time to try again" at once. The quarantine is our own
claim row's possible lifetime, the same thing the wait exists to outlast
for a crashed predecessor. So start() now waits it out even past the
wait, once, with time for a claim after it: at most about
start_lease_wait_s + 2 x lease_ttl_s against a catalog that slow, and
a second such timeout fails start() as before. (The design said to wait
the quarantine out "when it fits"; at the defaults it never fits.)

Or the server gave up first. The lease INSERTs carry the time the client
gives them as max_execution_time, lock_acquire_timeout and
insert_quorum_timeout, and ClickHouse answers a limit that ran out with
TIMEOUT_EXCEEDED (a 408), DEADLOCK_AVOIDED or UNKNOWN_STATUS_OF_INSERT.
execute() raised those as ordinary errors, so start() rethrew them, and
the service's lease-timeout count never saw them: on a catalog slow at
running the lease INSERT, the knobs were never named. They are now
timeouts, never retried, with messages that say which limit and which
bound; their outcome is as unknown as before, so a claim still
quarantines.

The conformance driver reports timed_out and sent for a ClickHouse
error, for the driver-level test.

On a fresh catalog the first lease request start() makes is not its own
claim but CatalogSchema::ensure()'s install-lease claim, on the same
coordinator. Each request of that claim now has the claim bound,
min(clickhouse_request_timeout_s, lease_ttl_s / 3) -- 5 s at the
defaults, where main gave it the 60 s request timeout -- and
take_the_install_lease caught only a refusal, so a cold server's first
lease read failed start(). The first start against a new catalog is
exactly when the server is likeliest to be cold, and the retry start()
now makes for its own claim never ran, since ensure() had already
thrown.

A timed-out install claim is now retried within the install lease's
existing budget (lease_ttl_s + 5 s), counted in time rather than in
attempts since one attempt can take a TTL. It is not quarantined, as it
was not before: an INSERT of it that lands late refuses the next claim
until it expires, which the refusal branch already waits out.

The slow-first-read start() test pre-warmed the schema so that the
service's own claim was the first lease read, and never exercised this;
it now runs against a fresh catalog too.

The service header said every use of the writer's lease goes through a
LeaseScope. ensure()'s install lease does not, before and after this
change; the comment now says so.

The lease-timeout streak, which names lease_ttl_s, clock_skew_s and
clickhouse_request_timeout_s once it reaches three, reset on every claim
that succeeded and counted only claims and renewals that timed out.
Against a catalog too slow to keep a lease the service goes round: a
claim goes through, then a renewal or a request of the index pass runs
out of lease, the writer quarantines, and the next claim goes through
again. The count went 1, 0, 1, 0, and the knobs were never named. A
renewal inside publish_snapshot, a pass request the lease deadline cut
off, and a lease abandoned at its deadline were never counted at all.

Every timeout that costs the lease now counts once:
- a claim or renewal that timed out, as before, the server's own time
  limit included;
- a lease abandoned at its deadline;
- any stretch under the lease lock that lost its lease to a request that
  timed out, which LeaseScope sees on its way out. The client records,
  in each RequestDeadline scope, whether the last request made under it
  timed out; a loss already counted inside the stretch is not counted
  again.

The count clears only once a lease has been held for 2 x TTL, as seen at
the end of a stretch under the lease lock -- not when a claim or a
renewal succeeds. (The design said "resets on a successful renewal";
against the catalog in the test the lease renews several times before a
pass request runs it out, so that would still hide the loop.)

The live test runs a service against a catalog answering INSERTs 0.9 s
and reads 0.4 s late at a 3 s TTL: claims go through and the lease is
lost after each, and the snapshot now names the knobs within about 15 s.

Tests:
- The in-pass stall test stalled only a catalog INSERT, which quarantines
  the writer by itself. It now also stalls the replay guard's read, which
  does not: the lease its deadline passed on is given up at the lease
  lock's scope, and the test checks it is never reported held over a
  dead row.
- A claim that wrote nothing (every connection refused) goes again a
  lease tick later, so that flush()'s fast cycles do not hammer a catalog
  that cannot answer. Nothing pinned it; the new test counts the
  connections flush() makes while the catalog refuses them (9 in 4 s,
  39 with the throttle removed).
- Two stall tests checked that the first "quarantined" sample came
  before 3.0 s, where the quarantine happens at about 2.9 s and samples
  are 0.1 s apart: a few milliseconds of slack. They now check when the
  writer actually quarantined, from the snapshot's quarantined_until,
  which is on time.monotonic()'s clock.

Comments: the lease tick is no longer what schedules a renewal -- the
lease thread wakes when one falls due, and a stretch under the lease lock
renews before its next request -- and renewal_window_ns is the floor the
configuration is checked against, not a promise that a renewal starts
within half the TTL.

#154's note on sweep_spool_on_start said a stalled holder keeps the lease
locally and starts new batches until a renewal or publish is refused. On
this branch a holder whose catalog requests stall has them cut off at the
lease deadline, which falls before its row lapses, and a cycle's lease
check abandons a lease past that deadline (LeaseScope), so it starts no
batch after it. Only a holder whose whole process stalls keeps the lease
locally past its row: until it resumes and checks again, or, after a
system suspend (the steady clock the deadline runs on does not count
one), until a renewal or publish is refused.